package edu.sysu.demo.rabbitmq.config;

import edu.sysu.demo.rabbitmq.serial.KryoMessageConverter;
import edu.sysu.demo.rabbitmq.serial.KryoSerializer;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

  @Bean
  KryoSerializer kryoSerializer() {
    return KryoSerializer.threadLocal();
  }

  @Bean
  Queue queueOdd() {
    return new Queue("cottle.odd");
  }

  @Bean
  Queue queueEven() {
    return new Queue("cottle.even");
  }

  @Bean
  TopicExchange topicExchange() {
    return new TopicExchange("amq.topic");
  }

  @Bean
  Binding bindingOdd(TopicExchange exchange) {
    return BindingBuilder.bind(queueOdd())
        .to(exchange)
        .with("cottle.odd.#");
  }

  @Bean
  Binding bindingEven(TopicExchange exchange) {
    return BindingBuilder.bind(queueOdd())
        .to(exchange)
        .with("cottle.even.#");
  }

  @Bean
  public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
      ConnectionFactory connectionFactory, KryoMessageConverter kryoMessageConverter) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setMessageConverter(kryoMessageConverter);
    return factory;
  }

  @Bean
  KryoMessageConverter kryoMessageConverter() {
    return new KryoMessageConverter();
  }
}
